package com.lkd.emq;

import com.lkd.exception.LogicException;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.List;

@Component
@Slf4j
public class MqttCallback implements MqttCallbackExtended {

    @Value("${mqtt.consumer.consumerTopics}")
    private List<String> topics;

    private MqttClient mqttClient;

    @Autowired
    private MqttService mqttService;

    public void setMqttClient(MqttClient mqttClient){
        this.mqttClient = mqttClient;
    }

    /**
     * 连接服务端成功
     * 订阅主题
     * @param reconnect
     * @param serverURI
     */
    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        log.info("连接成功，开始订阅主题");
        if(!CollectionUtils.isEmpty(topics)) {
            for (String topic : topics) {
                try {
                    log.info(">>>>>>>>>>>>>>订阅主题 topic=="+topic);
                    mqttClient.subscribe(topic, 2);
                } catch (MqttException e) {
                    log.error("订阅主题失败，topic="+topic);
                    e.printStackTrace();
                    throw new LogicException("订阅主题失败，topic="+topic);
                }
            }

        }
    }

    /**
     * 连接中断
     * @param cause
     */
    @Override
    public void connectionLost(Throwable cause) {
        log.error("EMQ的连接已经中断！");
    }



    /**
     * 消费者接收消息
     * @param topic
     * @param message
     * @throws Exception
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
//        log.info( "topic:"+topic+"  message:"+ new String(message.getPayload()));
        //处理消息
        mqttService.processMessage(topic, message);
        //处理成功后确认消息
        mqttClient.messageArrivedComplete(message.getId(),message.getQos());
    }

    /**
     * 发送消息成功后，回调方法
     * @param token
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        log.info("消息发送成功");
    }
}
